package com.dys.flink.wc2;

import com.dys.flink.wc1.util.JsonUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;

import java.time.Duration;
import java.util.Objects;

/**
 * 从 socket 9999 读 JSON 行，按 product_id 聚合，输出 AggResult
 */
public class SocketAggJob {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /* 1. 接 socket */
        SingleOutputStreamOperator<Order> orderStream = env
                .socketTextStream("localhost", 9999)
                .filter(line -> line != null && !line.trim().isEmpty()) // ① 过滤空行
                .map(line -> { // ② 捕获解析异常
                    try {
                        return JsonUtils.fromJson(line, Order.class);
                    } catch (Exception e) {
                        return null; // 解析失败返回 null
                    }
                })
                .filter(Objects::nonNull) // ③ 把 null 过滤掉
                .returns(Order.class);

        /* 2. 聚合 */
        SingleOutputStreamOperator<AggResult> aggStream = orderStream
                .keyBy(o -> o.product_id)
                .window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(10)))
                .aggregate(new OrderAggFunc());

        /* 3. 打印 */
        aggStream.print();

        env.execute("Socket-AggResult");
    }

    // ---------------- POJO ----------------
    public static class Order {
        public long id;
        public String product_name;
        public String product_id;
        public double product_price;
    }

    public static class AggResult {
        public String productId;
        public long   count;
        public double sumPrice;
        public double maxPrice;
        public double minPrice;
        public double avgPrice;

        public AggResult() {}

        public static AggResult fromFirst(Order o) {
            AggResult a = new AggResult();
            a.productId = o.product_id;
            a.count     = 1L;
            a.sumPrice  = o.product_price;
            a.maxPrice  = o.product_price;
            a.minPrice  = o.product_price;
            a.avgPrice  = o.product_price;
            return a;
        }

        @Override
        public String toString() {
            return String.format(
                    "AggResult{productId='%s', count=%d, sum=%.2f, max=%.2f, min=%.2f, avg=%.2f}",
                    productId, count, sumPrice, maxPrice, minPrice, avgPrice);
        }
    }

    public static class Acc {
        public String productId;
        public long   count = 0L;
        public double sum   = 0.0;
        public double max   = Double.MIN_VALUE;
        public double min   = Double.MAX_VALUE;
    }

    // ---------------- 聚合函数 ----------------
    public static class OrderAggFunc
            implements AggregateFunction<Order, Acc, AggResult> {

        @Override
        public Acc createAccumulator() {
            return new Acc();
        }

        @Override
        public Acc add(Order value, Acc acc) {
            if (acc.count == 0) {          // 第一条
                acc.productId = value.product_id;
            }
            acc.count++;
            acc.sum += value.product_price;
            acc.max  = Math.max(acc.max, value.product_price);
            acc.min  = Math.min(acc.min, value.product_price);
            return acc;
        }

        @Override
        public AggResult getResult(Acc acc) {
            AggResult r = new AggResult();
            r.productId = acc.productId;
            r.count     = acc.count;
            r.sumPrice  = acc.sum;
            r.maxPrice  = acc.max;
            r.minPrice  = acc.min;
            r.avgPrice  = acc.sum / acc.count;
            return r;
        }

        @Override
        public Acc merge(Acc a, Acc b) {
            a.count += b.count;
            a.sum   += b.sum;
            a.max    = Math.max(a.max, b.max);
            a.min    = Math.min(a.min, b.min);
            return a;
        }
    }
}